package cn.learn.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.AttributeKey;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;

public class NettyServerCnxnFactory extends ServerCnxnFactory {
    private final ServerBootstrap bootstrap;
    private InetSocketAddress localAddress;
    private Channel parentChannel;
    final ConcurrentHashMap<Long, NettyServerCnxn> sessionMap = new ConcurrentHashMap<Long, NettyServerCnxn>();
    private static final AttributeKey<NettyServerCnxn> CONNECTION_ATTRIBUTE = AttributeKey.valueOf("NettyServerCnxn");
    // 连接信道处理器
    CnxnChannelHandler channelHandler = new CnxnChannelHandler();

    public NettyServerCnxnFactory(){
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap().group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                // parent channel options 允许公用端口
                .option(ChannelOption.SO_REUSEADDR, true)
                // child channels options
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_LINGER, -1)
                .childHandler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
//                        pipeline.addLast(readIssuedTrackingHandler);
                        pipeline.addLast(channelHandler);
                    }
                });
        this.bootstrap = bootstrap;
    }

    @Override
    void configure(InetSocketAddress addr) throws IOException {
        localAddress = addr;
    }

    @Override
    protected void startup(ZooKeeperServer zkServer, boolean startServer) {
        // 开始
        start();
        setZooKeeperServer(zkServer);
        zkServer.startdata();    // 加载数据
        zkServer.startup();
    }

    @Override
    public void addSession(long sessionId, NettyServerCnxn cnxn) {
        sessionMap.put(sessionId, cnxn);
    }

    private void start() {
        parentChannel = bootstrap.bind(localAddress).syncUninterruptibly().channel();
        localAddress = (InetSocketAddress) parentChannel.localAddress();
    }

    class CnxnChannelHandler extends ChannelDuplexHandler {
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            super.write(ctx, msg, promise);
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            final Channel channel = ctx.channel();
            InetAddress addr = ((InetSocketAddress) channel.remoteAddress()).getAddress();
            System.out.println(addr.getHostAddress());
            NettyServerCnxn cnxn = new NettyServerCnxn(channel, zkServer, NettyServerCnxnFactory.this);
            ctx.channel().attr(CONNECTION_ATTRIBUTE).set(cnxn);
//            super.channelActive(ctx);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            super.channelInactive(ctx);
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            NettyServerCnxn cnxn = ctx.channel().attr(CONNECTION_ATTRIBUTE).get();
            cnxn.processMessage((ByteBuf) msg);
        }

        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.fireChannelReadComplete();
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            super.userEventTriggered(ctx, evt);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            super.exceptionCaught(ctx, cause);
        }
    }
}
